package com.tencent.oa.fm.cmp.core.task;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ScheduledFuture;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;

import com.tencent.oa.fm.cmp.core.platform.common.components.CommonConfigComponent;
import com.tencent.oa.fm.cmp.infrastructure.common.context.ServiceContext;
import com.tencent.oa.fm.jarvis.core.common.StringUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.stereotype.Component;

import com.tencent.oa.fm.cmp.core.notify.LockManager;
import com.tencent.oa.fm.cmp.core.notify.rule.TaskRule;

import lombok.extern.slf4j.Slf4j;

/**
 * 任务调度器
 *
 * @author fanglindeng
 */
@SuppressWarnings("rawtypes")
@Slf4j
@Component
public class TaskScheduler extends AutoRun implements ApplicationContextAware {
    @Autowired
    private TaskRuleRepository taskRuleRepository;

    private List<TaskRule> taskRules;

    private Set<ScheduledFuture> scheduledFutures = new HashSet<ScheduledFuture>();

    private ApplicationContext applicationContext;

    private volatile boolean inited; // spring是否初始化完成

    @Autowired
    private LockManager lockManager;

    @Autowired
    private CommonConfigComponent commonConfigComponent;

    @Value("${task.category:}")
    private String taskCategory;

    public static final String DEFAULT_TASK_CORE_PLUS_SIZE = "defaultTaskCorePlusSize";

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @PostConstruct
    public void init() {
        try {
            startTasks();
        } catch (Exception e) {
            log.error("init TaskScheduler error", e);
        }

        inited = true;
    }

    protected int getInitPoolSize() {
        taskRules = loadTaskRules();
        String defaultCoreSize = commonConfigComponent.getSystemConfigByKey(DEFAULT_TASK_CORE_PLUS_SIZE);
        if (StringUtils.isNotBlank(defaultCoreSize)) {
            try {
                return taskRules.size() + Integer.valueOf(defaultCoreSize);
            } catch (Exception e) {
                return taskRules.size() + 2;
            }
        }
        return taskRules.size() + 2;
    }

    public void startTasks() {
        if (taskRules == null || taskRules.size() == 0) {
            return;
        }

        for (TaskRule taskRule : taskRules) {
            String cron = taskRule.getCron();
            String taskClassName = taskRule.getTaskClassName();
            String taskMethodName = taskRule.getTaskMethodName();
            if (StringUtils.isBlank(cron) || StringUtils.isBlank(taskClassName)
                    || StringUtils.isBlank(taskMethodName)) {
                continue;
            }
            log.info("begin run taskRule the taskName is :'" + taskClassName + "' ");
            Class<?> taskClass;
            Method taskMethod;
            Object bean;
            try {
                taskClass = Class.forName(taskClassName);
                taskMethod = taskClass.getMethod(taskMethodName);
                taskMethod.setAccessible(true);
                bean = applicationContext.getBean(taskClass);
            } catch (Exception e) {
                log.error("start task error because some errors are in the rule, rule:" + taskRule, e);
                continue;
            }
            ScheduledFuture<?> scheduledFuture = threadPoolTaskScheduler.schedule(() -> {
                try {
                    // 允许并发执行
                    if (taskRule.isConcurrent()) {
                        invoke(taskMethod, bean, taskRule);
                    } else {
                        String lockCode = taskRule.getLockCode();
                        invokeWithLock(taskMethod, bean, lockCode, taskRule);
                    }
                } catch (Exception e) {
                    log.error("invoke task error, rule:" + taskRule, e);
                }
            }, new CronTrigger(cron));

            scheduledFutures.add(scheduledFuture);
        }
    }

    public void stopTasks() {
        if (scheduledFutures == null || scheduledFutures.size() == 0) {
            return;
        }

        for (ScheduledFuture scheduledFuture : scheduledFutures) {
            if (!scheduledFuture.isCancelled()) {
                scheduledFuture.cancel(false);
            }
        }

        scheduledFutures.clear();
    }

    private void invoke(Method taskMethod, Object bean, TaskRule taskRule)
            throws IllegalAccessException, IllegalArgumentException, InvocationTargetException {
        ServiceContext.current().getHeader().setTenantId(taskRule.getTenantId());
        taskMethod.invoke(bean);
    }

    private void invokeWithLock(Method taskMethod, Object bean, String lockCode, TaskRule taskRule)
            throws IllegalAccessException, IllegalArgumentException, InvocationTargetException {
        ServiceContext.current().getHeader().setTenantId(taskRule.getTenantId());
        boolean acquireLockSuccess = lockManager.acquireLock(lockCode, taskRule.getTenantId());
        //log.info("acquire lock '" + lockCode +  "' " + (acquireLockSuccess ? "success" : "fail"));

        if (acquireLockSuccess) {
            lockManager.startAutoLease(lockCode, taskRule.getTenantId());
            try {
                invoke(taskMethod, bean, taskRule);
            } catch (Exception e) {
                throw e;
            } finally {
                lockManager.cancelAutoLease(lockCode, taskRule.getTenantId());
                lockManager.releaseLock(lockCode, taskRule.getTenantId());
            }
        }
    }

    public List<TaskRule> loadTaskRules() {
        List<TaskRule> taskRules = taskRuleRepository.getAllRulesByType();
        if (StringUtil.isNullOrBlank(taskCategory)||taskCategory.equalsIgnoreCase("default")) {
            taskRules = taskRules.stream().filter(item -> StringUtil.isNullOrBlank(item.getTaskCategory())||item.getTaskCategory().equalsIgnoreCase("default")).collect(Collectors.toList());
        } else {
            taskRules = taskRules.stream().filter(item -> taskCategory.equalsIgnoreCase(item.getTaskCategory())).collect(Collectors.toList());
        }

        return taskRules;
    }

    @Override
    public void run() {
        if (inited) {
            List<TaskRule> newTaskRules = loadTaskRules();

            if (taskRulesChanged(taskRules, newTaskRules)) {
                if (taskRules.size() < newTaskRules.size()) {
                    threadPoolTaskScheduler.setPoolSize(newTaskRules.size());
                }
                taskRules = newTaskRules;

                stopTasks();

                startTasks();
            }
        }
    }

    private boolean taskRulesChanged(List<TaskRule> oldTaskRules, List<TaskRule> newTaskRules) {
        if (oldTaskRules == null || oldTaskRules.size() == 0 || newTaskRules == null || newTaskRules.size() == 0) {
            return true;
        }
        if (oldTaskRules.size() != newTaskRules.size()) {
            return true;
        }
        // 满足集合相等的定义
        if (oldTaskRules.containsAll(newTaskRules) && newTaskRules.containsAll(oldTaskRules)) {
            return false;
        }
        return true;
    }
}
